package com.kitten.comment.consumer;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.RandomUtil;
import com.github.phantomthief.collection.BufferTrigger;
import com.kitten.comment.constants.MQConstants;
import com.kitten.comment.constants.RedisKeyConstants;
import com.kitten.comment.domain.dataobject.CommentDO;
import com.kitten.comment.domain.mapper.CommentDOMapper;
import com.kitten.comment.enums.CommentLevelEnum;
import com.kitten.comment.model.dto.CountCommentPublishMqDTO;
import com.kitten.framework.common.utils.JsonUtils;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.assertj.core.util.Lists;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.TimeUnit;

/**
 * @author kitten
 */
@Slf4j
@Component
@RocketMQMessageListener(consumerGroup = "kitten_group_first_reply_comment_id" + MQConstants.TOPIC_COUNT_COMMENT,
        topic = MQConstants.TOPIC_COUNT_COMMENT
)
public class OneLevelCommentFirstReplyCommentIdUpdateConsumer implements RocketMQListener<String> {

    @Resource
    private RedisTemplate<String, Object> redisTemplate;

    @Resource
    private ThreadPoolTaskExecutor taskExecutor;

    @Resource
    private CommentDOMapper commentDOMapper;

    private BufferTrigger<String> bufferTrigger = BufferTrigger.<String>batchBlocking()
            .bufferSize(50000) // 缓存队列的最大容量
            .batchSize(1000)   // 一批次最多聚合 1000 条
            .linger(Duration.ofSeconds(1)) // 多久聚合一次（1s 一次）
            .setConsumerEx(this::consumeMessage) // 设置消费者方法
            .build();

    @Override
    public void onMessage(String s) {
        bufferTrigger.enqueue(s);
    }

    private void consumeMessage(List<String> messages) {
        log.info("==> 【评论热度值计算】聚合消息, size: {}", messages.size());
        log.info("==> 【评论热度值计算】聚合消息, {}", JsonUtils.toJsonString(messages));

        List<CountCommentPublishMqDTO> mqDTOList = Lists.newArrayList();
        messages.forEach(message -> {
            try {
                List<CountCommentPublishMqDTO> list = JsonUtils.parseList(message, CountCommentPublishMqDTO.class);
                mqDTOList.addAll(list);
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        });
        // 过滤所有的二级评论, 获取这些评论的 父评论ID
        List<Long> parentIds = mqDTOList.stream().filter(dto -> Objects.equals(dto.getLevel(), CommentLevelEnum.TWO.getCode()))
                .map(CountCommentPublishMqDTO::getParentId)
                .distinct()
                .toList();
        if (CollUtil.isEmpty(parentIds)) {
            return;
        }
        // 构建Redis Key
        List<String> keys = parentIds.stream().map(RedisKeyConstants::buildHaveFirstReplyCommentKey).toList();
        // 批量查询
        List<Object> values = redisTemplate.opsForValue().multiGet(keys);
        // 提取 Redis中不存在的评论 ID
        List<Long> notExistsCommentIds = Lists.newArrayList();
        for (int i = 0; i < values.size(); i++) {
            if (Objects.isNull(values.get(i))) {
                notExistsCommentIds.add(parentIds.get(i));
            }
        }
        if (CollUtil.isNotEmpty(notExistsCommentIds)) { // Redis中不存在comment:haveFirstReplyCommentId：xxx的, 需要通过查询数据库确定是否有first_reply_comment_id
            List<CommentDO> commentDOS = commentDOMapper.selectByCommentIds(notExistsCommentIds);
            taskExecutor.submit(() -> {
                // 过滤出有first_reply_comment_id的评论
                List<Long> needSync2RedisCommentIds = commentDOS.stream().filter(commentDO -> commentDO.getFirstReplyCommentId() != 0)
                        .map(CommentDO::getId)
                        .toList();
                sync2Redis(needSync2RedisCommentIds);
            });
            // 更新一级评论的 first_reply_comment_id
            // 过滤出值为0的, 更新其 first_reply_comment_id
            List<CommentDO> needUpdateCommentDOS = commentDOS.stream().filter(commentDO -> commentDO.getFirstReplyCommentId() == 0).toList();

            needUpdateCommentDOS.forEach(commentDO -> {
                Long needUpdateCommentId = commentDO.getId();
                CommentDO earliestCommentDO = commentDOMapper.selectEarliesByParentId(needUpdateCommentId);
                if (Objects.nonNull(earliestCommentDO)) {
                    Long earliestCommentDOId = earliestCommentDO.getId();
                    commentDOMapper.updateFirstReplyCommentIdByPrimaryKey(earliestCommentDOId, needUpdateCommentId);
                    taskExecutor.submit(() -> {
                        // 同步到Redis
                        sync2Redis(Lists.newArrayList(needUpdateCommentId));
                    });
                }
            });

        }


    }

    // 被首次回复(拥有首次回复)的一级评论id, 同步到redis 记录并设置过期时间; RedisKey: comment:haveFirstReplyCommentId:xxx
    private void sync2Redis(List<Long> needSync2RedisCommentIds) {
        ValueOperations<String, Object> stringObjectValueOperations = redisTemplate.opsForValue();
        // 使用Redis的管道模式, 在一个操作中发送多个命令
        redisTemplate.executePipelined((RedisCallback<?>) connection -> {
           needSync2RedisCommentIds.forEach(commentId -> {
               String key = RedisKeyConstants.buildHaveFirstReplyCommentKey(commentId);
               // 设置过期时间, 随机5小时内
               stringObjectValueOperations.set(key, 1, RandomUtil.randomInt(5 * 60 * 60), TimeUnit.SECONDS);
           });
           return null;
        });
    }


}
